package com.haruhi.bot.endpoint.ws;

import com.haruhi.bot.endpoint.ws.model.MessageBox;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public class Consumer {

    private static final AtomicBoolean started = new AtomicBoolean(false);

    private static ExecutorService executor;

    public synchronized static void start(Collection<MessageHandler> handlers){
        initPool(handlers.size() > 0 ? handlers.size() : 1);
        if(!started.get()){
            started.set(true);
            new Thread(()->{
                while (true){
                    try {
                        MessageBox messageBox = MessageQueue.take();
                        for (MessageHandler handler : handlers) {
                            executor.execute(()-> {
                                try {
                                    handler.handle(messageBox);
                                } catch (Exception e) {
                                    log.error("处理消息异常 class:{} message:{}",handler,messageBox.getMessage(),e);
                                }
                            });
                        }
                    } catch (InterruptedException ignored) {

                    }
                }
            }).start();
        }
    }

    private synchronized static void initPool(int poolSize){
        if(executor == null){
            executor = new EnhancedThreadPoolExecutor(poolSize, poolSize * 2, 60,
                    TimeUnit.SECONDS, false, new CustomizableThreadFactory("P-MH-"));
        }
    }


    /**
     * 常规线程池执行任务规则： corePoolSize -> queueSize -> maximumPoolSize -> RejectedExecutionHandler
     *      corePoolSize未满，且有空闲线程，也会创建新线程
     * 该线程池执行任务规则： corePoolSize -> maximumPoolSize -> queueSize -> RejectedExecutionHandler
     *      corePoolSize未满，且有空闲线程，不会创建新线程，而是使用空闲线程去执行任务
     */
    static class EnhancedThreadPoolExecutor implements ExecutorService {

        private final ThreadPoolExecutor delegate;

        EnhancedThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, boolean allowCoreThreadTimeOut, ThreadFactory threadFactory){
            delegate = new ThreadPoolExecutor(corePoolSize,maximumPoolSize,keepAliveTime,unit,new Queue(this),threadFactory);
            delegate.allowCoreThreadTimeOut(allowCoreThreadTimeOut);
        }

        public int getActiveCount(){
            return delegate.getActiveCount();
        }
        public int getPoolSize(){
            return delegate.getPoolSize();
        }
        public BlockingQueue<Runnable> getQueue(){
            return delegate.getQueue();
        }
        public int getCorePoolSize(){
            return delegate.getCorePoolSize();
        }
        public int getMaximumPoolSize(){
            return delegate.getMaximumPoolSize();
        }


        @Override
        public void execute(Runnable runnable){
            if(getActiveCount() < getPoolSize()){
                if(getQueue().offer(runnable)){
                    return;
                }
            }
            delegate.execute(runnable);
        }

        @Override
        public void shutdown() {
            delegate.shutdown();
        }

        @Override
        public List<Runnable> shutdownNow() {
            return delegate.shutdownNow();
        }

        @Override
        public boolean isShutdown() {
            return delegate.isShutdown();
        }

        @Override
        public boolean isTerminated() {
            return delegate.isTerminated();
        }

        @Override
        public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
            return delegate.awaitTermination(timeout,unit);
        }

        @Override
        public <T> Future<T> submit(Callable<T> task) {
            return delegate.submit(task);
        }

        @Override
        public <T> Future<T> submit(Runnable task, T result) {
            return delegate.submit(task,result);
        }

        @Override
        public Future<?> submit(Runnable task) {
            return delegate.submit(task);
        }

        @Override
        public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
            return delegate.invokeAll(tasks);
        }

        @Override
        public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
            return delegate.invokeAll(tasks,timeout,unit);
        }

        @Override
        public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
            return delegate.invokeAny(tasks);
        }

        @Override
        public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
            return delegate.invokeAny(tasks,timeout,unit);
        }
    }

    static class Queue extends LinkedBlockingQueue<Runnable> {
        private final EnhancedThreadPoolExecutor parent;
        Queue(EnhancedThreadPoolExecutor executor){
            this.parent = executor;
        }
        @Override
        public boolean offer(Runnable runnable) {
            if (parent.getPoolSize() < parent.getMaximumPoolSize()) {
                return false;
            }
            return super.offer(runnable);
        }
    }
}
